KAFKA-19893: Reduce tiered storage redundancy with delayed upload (KIP-1241) by jiafu1115 · Pull Request #20913 · apache/kafka

jiafu1115 · 2025-11-18T11:23:27Z

JIRA:19893
KIP:1241

Currently, Kafka uploads all non-active local log segments to remote
storage even when they are still within the local retention period,
resulting in redundant storage of the same data in both tiers.

This wastes storage capacity (cost) without providing immediate
benefits,since reads during the retention window prioritize local data.

However, some users/topics do real-time analytics based on remote
storage directly and need the latest data to be available as soon as
possible (In fact, it only tries to stay as up-to-date as possible, but
it still can’t include the latest data because the active segment
hasn’t been uploaded yet.). Therefore, this optimization is offered as
a topic's optional configuration rather than the default behavior.

Here are some additional thoughts/considerations.

Local files won’t be deleted until they’ve been uploaded to the
remote storage, so this change is very safe—you don’t need to worry
about files being cleaned up before they be upload to the remote.
Considering the latency of remote storage, the local retention period
won’t be set too short. For example, in our production environment, we
keep 1 day of local data alongside 3-7 days in remote storage, so
there’s still 1 day of redundancy.

Example for the goal:

Reviewers: Kamal Chandraprakash kamal.chandraprakash@gmail.com

Signed-off-by: stroller <fujian1115@gmail.com>

jiafu1115 · 2025-11-18T11:36:37Z

Attach test result:
[Precondition]
Create one topic enable remote stroage in Kafka (3 brokers + 3 controller)

local storage time: 20 minutes
remote stroage time: 40 minutes
partition:  3
segement.bytes: 10M

[Steps]

Deploy this code patch into one broker only and restart the broker
Keep sending a lot of messages to the topic
Check the disk sizes on both local and remote storage at two points in time: 20 minutes before and 1 hour after.

[Result]

Before 20 minutes:

only 2 partition upload the local to remote.

After 1 hour:

The remote storage size for one partition (on the broker with the code change) is much smaller than the other two.

2. The sizes of the local disks are similar.

… for remote storage. Signed-off-by: Jian <fujian1115@gmail.com>

Signed-off-by: Jian <fujian1115@gmail.com>

github-actions · 2025-11-26T03:29:39Z

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

jiafu1115 · 2025-11-28T02:09:13Z

Hi, @kamalcph
Sorry to bother you. I know you’ve been deeply involved in the remote storage area, and I was wondering if you might be interested — when you have some free time — in taking a look at this cost-saving topic and providing some guidance. Thank you very much!

github-actions · 2025-11-29T03:27:23Z

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

jiafu1115 · 2025-12-02T10:17:14Z

cc @kamalcph here due to community's email don't allow to attach the image. We can discuss the content in email about the KIP. Thanks

kamalcph · 2025-12-02T11:23:51Z

@jiafu1115

The already uploaded segments are eligible for deletion from broker. So, when remote storage is down, then those segments can be deleted as per the local retention settings and new segments can occupy those space. This provides more time for the Admin to act when remote storage is down for a longer time.

jiafu1115 · 2025-12-02T12:21:53Z

@kamalcph I think I understand what you mean now. I’ve updated the picture above. Could you help double-check whether we’ve reached the same understanding?
The drawback of this KIP is that, during a long time remote storage outage. it will occupied more disk so that admin may need one extra disk expansion. The max value is the redundant part we saving.
Thus. After the outage recovered. It will come back to the beginning.
Right?

github-actions · 2025-12-04T03:33:27Z

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

github-actions · 2025-12-06T03:26:03Z

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

jiafu1115 · 2026-05-18T05:36:54Z

@kamalcph

Hi Kamal:
Sorry to trouble you. I refactored the code based on your suggestion.
0 is the default value only mean upload at once only. -1 is the max lag value as local retention time/size.

It seems the logic is a bit simpler now and the concepts are more clearer, and it’s not as complicated as I originally thought. so I decided to go with your approach. Could you help review it again?

I pasted the test matrix here for your reference to help check the logic.

kamalcph

Thanks @jiafu1115 for addressing the review comments. We may have to further simplify the logic.

The current logic has multiple negative conditions, which makes it harder to follow. The delayCopy method internally calls notExceededCopyLagTime and notExceededCopyLagSize. Both of these not... methods check whether the threshold is exceeded and then invert the result again. Can we update this logic to use a positive flow instead?

(eg)

List<EnrichedLogSegment> candidateLogSegments(UnifiedLog log, Long fromOffset, Long lastStableOffset) {
...
...
 if (isEligibleForUpload(log.config(), previousSeg, currentTimeMs, totalLogSize, cumulativeSize)) {
      candidateLogSegments.add(new EnrichedLogSegment(previousSeg, currentSeg.baseOffset()));
  } else {
      break;
  }

and update the delay logic to:

private boolean isEligibleForUpload(LogConfig logConfig,
                                LogSegment previousSeg,
                                long currentTimeMs,
                                long totalLogSize,
                                long cumulativeSize) {
  long effectiveCopyLagMs = logConfig.remoteCopyLagMs();
  // derive the value from local retention time when copyLagMs configured to -1
  if (effectiveCopyLagMs == -1L) {
      effectiveCopyLagMs = logConfig.localRetentionMs();
      // if the local retention time is configured to infinite, then configure copyLagMs as infinite
      if (effectiveCopyLagMs == -1L) {
          effectiveCopyLagMs = Long.MAX_VALUE;
      }
  }
  long effectiveCopyLagBytes = logConfig.remoteCopyLagBytes();
  // derive the value from local retention size when copyLagBytes configured to -1
  if (effectiveCopyLagBytes == -1L) {
      effectiveCopyLagBytes = logConfig.localRetentionBytes();
      // if the local retention size is configured to infinite, then configure copyLagBytes as infinite
      if (effectiveCopyLagBytes == -1L) {
          effectiveCopyLagBytes = Long.MAX_VALUE;
      }
  }
  
  try {
      long segmentAgeMs = currentTimeMs - previousSeg.largestTimestamp();
      // If the segment's largestTimestamp is higher than the current time, then allow the segment to upload.
      boolean shouldUploadNow = segmentAgeMs < 0;
      // Upload when either of remote-copy lag time/size breaches the threshold.
      if (!shouldUploadNow) {
          shouldUploadNow = segmentAgeMs >= effectiveCopyLagMs;
      }
      if (!shouldUploadNow) {
          long sizeLagBytes = totalLogSize - cumulativeSize;
          shouldUploadNow = sizeLagBytes >= effectiveCopyLagBytes;
      }
      return shouldUploadNow;
  } catch (IOException e) {
      // in case of any error, allow the segment to upload. Should not block the upload that might hinder the
      // deletion logic
      LOGGER.warn("Failed to get largest timestamp for segment {}, marking it as eligible for upload based on time", previousSeg, e);
      return true;
  }
}

I ran the newly added unit tests and it is passing.

Also, move the unit tests from RemoteLogManagerTest to a new RemoteLagCopyTest since the RemoteLogManagerTest is huge with 4k+ lines.

Signed-off-by: stroller.fu <stroller.fu@zoom.us>

Signed-off-by: Jian <fujian1115@gmail.com>

Signed-off-by: stroller.fu <stroller.fu@zoom.us>

Signed-off-by: Jian <fujian1115@gmail.com>

kamalcph

LGTM, thanks for addressing the review comments! Left few minor comments.

kamalcph · 2026-05-22T07:03:11Z

  }

+  @Test
+  def testDynamicRemoteCopyLagThrowsOnIncorrectConfig(): Unit = {


could you also add one test for valid dynamic broker config change? Thanks!

Also, move the testDynamicRemoteCopyLagThrowsOnIncorrectConfig test to DynamicBrokerConfigTest.java instead of DynamicBrokerConfigTest.scala.

This can be taken up later, we already have one dynamic broker config change test in KafkaConfigTest.

Got it. thanks @kamalcph
Thank you very much for your patient and thorough review.

chia7712

@jiafu1115 thanks for this patch. It is cooool

chia7712 · 2026-05-22T15:52:26Z

+        private boolean eligibleUploadByTime(LogSegment segment, long currentTimeMs, long copyLagMs) {
+            try {
+                long segmentAgeMs = currentTimeMs - segment.largestTimestamp();
+                boolean eligibleUpload = segmentAgeMs < 0 || segmentAgeMs >= copyLagMs;


This is a lingering issue. The local segment with a future timestamp is still NOT deleted, right? Should we allow the deletion of these local files once they have been successfully uploaded to remote storage, even if they contain future timestamps?

Should we allow the deletion of these local files once they have been successfully uploaded to remote storage, even if they contain future timestamps?

Currently, it gets deleted based on the local retention bytes limit since the local retention deletion logic follow the similar methods of full deletion logic.

https://sourcegraph.com/r/github.com/apache/kafka/-/blob/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java?L2008

The current behavior looks like a bug to me, as local segments shouldn't be blocked from deletion once they are already uploaded. Perhaps we could fall back to checking the lastModifiedTime when a segment contains future records. Alternatively, we could introduce a new configuration flag to provide alternative logic for handling future records.

shouldn't be blocked from deletion once they are already uploaded. Perhaps we could fall back to checking the lastModifiedTime when a segment contains future records.

Yeah, we can allow this behavior by introducing a config. The segment exist in the remote storage but the user might face slowness in reading the data from remote if they don't have prefetching feature implemented in the Remote Storage Manager. So, better to gate the change in behavior via config and the change applies only when remote storage is enabled on the topic.

If we all hate -2, there is another approach: reverse the configuration logic.

For example, we could introduce remote.copy.before.deletion.ms instead of remote.copy.lag.ms. If a user sets remote.copy.before.deletion.ms=2hr, it means they want to upload log segments 2 hours before they are scheduled to be deleted from local storage.

The default value would be -1, which dynamically resolves to local.retention.ms. This fully maintains backward compatibility (resulting in immediate upload). Setting it to 0 means the user wants to delay the upload as much as possible, triggering it right before local deletion.

The only side effect is that this kind of "reverse" or "countdown" time calculation is quite rare in the existing Kafka codebase.

The only side effect is that this kind of "reverse"

yeah, this is not intuitive when compared with the other configs like retention time/bytes. My suggestion is to:

Improve the config documentation about configuring both the values and

Add validation to throw an error when remote-copy-lag-time = 0 and remote-copy-lag-bytes != 0 and vice-versa.

I just realized that I replied to the wrong thread earlier, sorry about that 😢

How about this solution? if we found it contain future record. we use LogSegment#lastModified to compare the time?

Yes, that is an acceptable approach. I've opened https://issues.apache.org/jira/browse/KAFKA-20609 so we can keep discussing there.

yeah, this is not intuitive when compared with the other configs like retention time/bytes. My suggestion is to:

While documentation is the final line of defense for users, it's better to have an intuitive design out of the box. After all, me proposed reverse configuration was rejected precisely because it wasn't intuitive enough

Another way is to align the logic with retention.ms and retention.bytes. Since most users care more about time than size, we could set the default value of the time lag to 0, and the size lag to -1. This way, most users can just adjust the time setting without having to touch the size configuration. WDYT?

Thanks for filing KAFKA-20609 ticket. Fixing this is out of scope of the KIP-1241 as the bug exist before too.

Since most users care more about time than size, we could set the default value of the time lag to 0, and the size lag to -1.

I like the idea to provide out-of-box default values that works in majority of use-cases and retain the eager upload logic. The new default values for remote-copy lag align with the log.retention.hours = 7 days and log.retention.bytes = infinite (-1). It lgtm.

@kamalcph @chia7712 ack
Let me take time to think and handle it. Thanks

Opened #22394 to address this comment. PTAL.

chia7712 · 2026-05-22T15:53:27Z

        }
    }

+    private static void validateRemoteCopyLagTime(Map<?, ?> props) {


Map<?, ?> ->Map<String, ?>

duplicated with another one. propose the fix in #22363 . thanks.

chia7712 · 2026-05-22T15:53:31Z

+        }
+    }
+
+    private static void validateRemoteCopyLagSize(Map<?, ?> props) {


Map<?, ?> ->Map<String, ?>

@chia7712 thanks for review. Fix it on #22363

chia7712 · 2026-05-22T16:03:22Z

+                        previousSeg, copyLagMs, copyLagBytes, currentTimeMs, totalLogSize, cumulativeSize, totalLogSize - cumulativeSize);
+            }
+
+            if (copyLagMs == 0 || copyLagBytes == 0) {


Should we log a warning or info message when a user configures one lag property but leaves the other at its default value of 0?

@chia7712 Thanks for your view.
How about describing this case in the documentation? If we only log it as a warning, users may ignore it. WDYT?

Sounds good. We can expand upgrade.md as well.

@kamalcph

This is a follow-up PR for #20913 (comment) Thanks. cc @kamalcph Reviewers: Murali Basani <muralidhar.basani@aiven.io>, Ken Huang <s7133700@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>

Reduce remote storage by checking local retention

cb6f22f

Signed-off-by: stroller <fujian1115@gmail.com>

github-actions Bot added triage PRs from the community storage Pull requests that target the storage module tiered-storage Related to the Tiered Storage feature clients small Small PRs labels Nov 18, 2025

jiafu1115 marked this pull request as ready for review November 18, 2025 11:37

jiafu1115 changed the title ~~KAFKA-19893: Reduce redundant storage in remote tier (configurable)~~ KAFKA-19893: Reduce redundant storage in remote tier Nov 18, 2025

jiafu1115 changed the title ~~KAFKA-19893: Reduce redundant storage in remote tier~~ KAFKA-19893: Reduce redundant storage in remote tier (configurable) Nov 19, 2025

chia7712 added the ci-approved label Nov 19, 2025

jiafu1115 changed the title ~~KAFKA-19893: Reduce redundant storage in remote tier (configurable)~~ KAFKA-19893: Reduce tiered storage redundancy with delayed upload Nov 19, 2025

jiafu1115 changed the title ~~KAFKA-19893: Reduce tiered storage redundancy with delayed upload~~ KAFKA-19893: Reduce tiered storage redundancy with delayed upload (KIP-1241) Nov 19, 2025

jiafu1115 added 2 commits November 19, 2025 20:45

correct the configure name to keep same style with existed configures…

4f96275

… for remote storage. Signed-off-by: Jian <fujian1115@gmail.com>

fix the unit test for null point protect

c410ae3

Signed-off-by: Jian <fujian1115@gmail.com>

github-actions Bot added the needs-attention label Nov 26, 2025

github-actions Bot removed the needs-attention label Nov 28, 2025

github-actions Bot added the needs-attention label Nov 29, 2025

github-actions Bot removed the needs-attention label Dec 3, 2025

github-actions Bot added the needs-attention label Dec 4, 2025

jiafu1115 changed the title ~~KAFKA-19893: Reduce tiered storage redundancy with delayed upload (KIP-1241)~~ KAFKA-19893: Reduce tiered storage redundancy with delayed upload (Topic-level feature) (KIP-1241) Dec 4, 2025

github-actions Bot removed the needs-attention label Dec 5, 2025

kamalcph reviewed May 18, 2026

View reviewed changes

Comment thread storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java Outdated

jiafu1115 and others added 4 commits May 18, 2026 15:40

refactor the unit test: split to dedicated test class

5880721

Signed-off-by: stroller.fu <stroller.fu@zoom.us>

remove the unrelated code change.

8f23a7c

Signed-off-by: stroller.fu <stroller.fu@zoom.us>

simpler the code

6d218ff

Signed-off-by: stroller.fu <stroller.fu@zoom.us>

refactor the code

d478387

Signed-off-by: stroller.fu <stroller.fu@zoom.us>

jiafu1115 commented May 18, 2026

View reviewed changes

Comment thread storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java

jiafu1115 requested a review from kamalcph May 18, 2026 14:10

jiafu1115 added 7 commits May 19, 2026 04:14

add one unit test for both lag and retention is -1

134337d

Signed-off-by: Jian <fujian1115@gmail.com>

reduce the code complex

499cb59

Signed-off-by: Jian <fujian1115@gmail.com>

continue to reduce the code complex

cbf725c

Signed-off-by: Jian <fujian1115@gmail.com>

continue to reduce the code complex

40d91a7

Signed-off-by: Jian <fujian1115@gmail.com>

continue to reduce the code complex

20ca059

Signed-off-by: Jian <fujian1115@gmail.com>

correct the unit test

8f4cdd7

Signed-off-by: stroller.fu <stroller.fu@zoom.us>

remove the document for -1

3f0764f

Signed-off-by: stroller.fu <stroller.fu@zoom.us>

kamalcph reviewed May 21, 2026

View reviewed changes

jiafu1115 added 4 commits May 21, 2026 19:43

simple the code according to the code review suggestion

a8ad430

Signed-off-by: Jian <fujian1115@gmail.com>

handle the special case for segment age.

367390a

Signed-off-by: Jian <fujian1115@gmail.com>

add more unit test

dbaa4d6

Signed-off-by: Jian <fujian1115@gmail.com>

Merge branch 'apache:trunk' into storage

9cfbc5e

jiafu1115 requested a review from kamalcph May 21, 2026 15:53

kamalcph reviewed May 22, 2026

View reviewed changes

kamalcph approved these changes May 22, 2026

View reviewed changes

kamalcph merged commit edcada2 into apache:trunk May 22, 2026
31 checks passed

This was referenced May 22, 2026

MINOR: Correct the variable name related to remote delay storage in the test #22348

Closed

MINOR: Correct the variable name in remote storage's test #22349

Merged

chia7712 reviewed May 22, 2026

View reviewed changes

jiafu1115 mentioned this pull request May 24, 2026

MINOR: Narrow remote copy lag validator props to Map<String, ?> and update the upgrade doc for KIP-1241 #22363

Open

kamalcph mentioned this pull request May 28, 2026

KAFKA-20633: Update the default value of remote copy lag bytes #22394

Open

Conversation

jiafu1115 commented Nov 18, 2025 • edited by github-actions Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jiafu1115 commented Nov 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions Bot commented Nov 26, 2025

Uh oh!

jiafu1115 commented Nov 28, 2025

Uh oh!

github-actions Bot commented Nov 29, 2025

Uh oh!

jiafu1115 commented Dec 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

kamalcph commented Dec 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jiafu1115 commented Dec 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions Bot commented Dec 4, 2025

Uh oh!

github-actions Bot commented Dec 6, 2025

Uh oh!

jiafu1115 commented May 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

kamalcph left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

kamalcph left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kamalcph May 25, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kamalcph May 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jiafu1115 commented Nov 18, 2025 •

edited by github-actions Bot

Loading

jiafu1115 commented Nov 18, 2025 •

edited

Loading

jiafu1115 commented Dec 2, 2025 •

edited

Loading

kamalcph commented Dec 2, 2025 •

edited

Loading

jiafu1115 commented Dec 2, 2025 •

edited

Loading

jiafu1115 commented May 18, 2026 •

edited

Loading

kamalcph left a comment •

edited

Loading

kamalcph May 25, 2026 •

edited

Loading

kamalcph May 26, 2026 •

edited

Loading